package com.rainman.simpleredis.core.scheduled;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.rainman.simpleredis.core.config.SimpleRedisConfig;
import com.rainman.simpleredis.core.distributed_lock.SimpleDistributedLock;
import com.rainman.simpleredis.core.utils.CronSequenceGenerator;

import java.time.Duration;
import java.util.Date;
import java.util.HashMap;
import java.util.Set;
import java.util.TimeZone;


/**
 * 定时任务底层依赖于redis的键过期
 * 因为键过期监听会有延迟(过期机制有延迟，网络传输有延迟)
 * 对于时间要求准确度要求及其严格的业务谨慎使用
 *
 */
public class SimpleScheduled {
    private final SimpleRedisConfig simpleRedisConfig;
    private final SimpleDistributedLock simpleDistributedLock;
    //解析corn
    private final HashMap<String, CronSequenceGenerator> cronSequenceGeneratorHashMap;
    //业务头数据
    public  final String ScheduledOri;//SR0.Ori.Sch:
    public  final String ScheduledTig;//SR0.Tig.Sch:
    //分布式锁头数据
    public  final String ScanDataLockHead="SimpleRedis.SchScanData";
    public  final String ScanDataDealingLockHead="SimpleRedis.SchScanDataDealing";
    public  final String DealLockHead="SimpleRedis.SchDealOne:";

    public SimpleScheduled(SimpleRedisConfig simpleRedisConfig ,SimpleDistributedLock simpleDistributedLock){
        this.simpleRedisConfig = simpleRedisConfig;
        this.simpleDistributedLock = simpleDistributedLock;
        cronSequenceGeneratorHashMap=new HashMap<>();
        String keyHead=simpleRedisConfig.getKeyHead()+ simpleRedisConfig.getProduct()+".";//SR0.

        ScheduledOri =keyHead+"Ori.Sch:";//SR0.Ori.Sch:
        ScheduledTig =keyHead+"Tig.Sch:";//SR0.Tig.Sch:

        new Thread(()-> {
                try {
                    Thread.sleep(300L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                scanData();

                while (true){

                    try {
                        Thread.sleep(33555L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                    try{
                        Boolean aBoolean = simpleDistributedLock.dLockAdd(ScanDataLockHead,33333L);
                        if(aBoolean){
                            scanData();
                        }
                    }catch (Exception e){
                        e.printStackTrace();
                    }

                }
        }).start();

    }


    /**
     * 兜底扫描数据
     */
    private void scanData() {

        Boolean aBoolean = simpleDistributedLock.dLockAdd(ScanDataDealingLockHead,40000L);
        if(!aBoolean){
            return;
        }

        Set<String> oriInfoKeys = simpleRedisConfig.getRedisInterface().keys(ScheduledOri + "*");//SR0.Ori.Sch:*

        for(String key:oriInfoKeys){
            String[] split = key.split(":");//SR0.Ori.Sch:oriKey
            String oriKey=split[1];

            //----------------------------------------------------------------------
            Boolean hasNextTig = simpleRedisConfig.getRedisInterface().hasKey(ScheduledTig + oriKey);
            //检测是否已有触发
            if(hasNextTig){continue;}

            Boolean duringDeal = simpleDistributedLock.hasLock(DealLockHead + oriKey);
            //检测是否进行中
            if(duringDeal){continue;}

            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //再检测一次。某些情况会存在触发的过期键已过期但是未执行（redis过期键监听的特性）

            hasNextTig = simpleRedisConfig.getRedisInterface().hasKey(ScheduledTig + oriKey);
            //检测是否已有触发
            if(hasNextTig){continue;}

            //检测是否进行中
            duringDeal = simpleDistributedLock.hasLock(DealLockHead + oriKey);
            //检测是否进行中
            if(duringDeal){continue;}

            //----------------------------------------------------------------------

            String value = simpleRedisConfig.getRedisInterface().getValue(key);

            JSONObject msgObject = JSONUtil.parseObj(value);
            Long schDur = getSchDur(msgObject);

            if(schDur>1000L*3600){//大于半小时的先不处理，给下一个兜底任务处理
                continue;
            }

            System.out.println("定时任务兜底触发"+ ScheduledTig + oriKey);

            //兜底3次后删除原始信息
            Integer count = msgObject.getInt("count");
            if(count<1){
                simpleRedisConfig.getRedisInterface().delete(key);
                continue;
            }

            msgObject.set("count",count-1);
            simpleRedisConfig.getRedisInterface().set(key,msgObject.toString());

            //插入触发
            simpleRedisConfig.getRedisInterface().set(ScheduledTig + oriKey,"", Duration.ofMillis(schDur));

        }

        simpleDistributedLock.dLockRemove(ScanDataDealingLockHead);
    }


    /**
     * 处理redis键过期
     * 即触发执行
     *
     * @param key
     */
    public void dealOne(String key){

        if(key==null){
            return;
        }
        String oriKey=key.split(":")[1];
        SimpleScheduledWorkerInterface workerInterface = simpleRedisConfig.getScheduledWorkerInterfacesMap().get(oriKey);
        if(workerInterface==null){//没有处理函数 不抢
            return;
        }

        //利用分布式锁，抢执行权
        Boolean aBoolean =simpleDistributedLock.dLockAdd(DealLockHead + oriKey,40000L);
        if(!aBoolean){
            return;
        }

        String value = simpleRedisConfig.getRedisInterface().getValue(ScheduledOri + oriKey);

        if(value==null){
            return;
        }

        JSONObject jsonObject = JSONUtil.parseObj(value);

        if(jsonObject.getInt("count")<3){
            jsonObject.set("count",3);
            simpleRedisConfig.getRedisInterface().set(ScheduledOri + oriKey,jsonObject.toString());
        }

        Long fixedRate=jsonObject.getLong("rate",0L);

        if(fixedRate>0L){
            //直接加下一个触发

            simpleDistributedLock.dLockRemove(DealLockHead + oriKey);
            simpleRedisConfig.getRedisInterface().set(ScheduledTig +oriKey,"", Duration.ofMillis(fixedRate));
        }

        try{

            workerInterface.doWork();

        }finally {

            if(fixedRate<=0){//加入下一个tig

                Long dur = getSchDurWithOutRate(jsonObject);

                simpleDistributedLock.dLockRemove(DealLockHead + oriKey);
                simpleRedisConfig.getRedisInterface().set(ScheduledTig +oriKey, "", Duration.ofMillis(dur));

            }

        }

    }

    /**
     * 新增动态任务
     *
     *
     * @param workerInterface
     */
    public void addWorker(SimpleScheduledWorkerInterface workerInterface){

        simpleRedisConfig.getScheduledWorkerInterfacesMap().put(workerInterface.key(),workerInterface);

        JSONObject jsonObject = JSONUtil.parseObj(workerInterface.info());
        jsonObject.set("count",3);

        simpleRedisConfig.getRedisInterface().set(ScheduledOri +workerInterface.key(), jsonObject.toString());

        //加入触发
        Long schDur = getSchDur(jsonObject);
        if(schDur<=1000L*3600){//大于半小时的先不处理，给兜底任务处理
            simpleRedisConfig.getRedisInterface().set(ScheduledTig + workerInterface.key(),"", Duration.ofMillis(schDur));
        }

    }

    /**
     * 去除任务
     * 删除任务源数据
     * 所有终端都不会再触发任务
     * 注意
     *  有包含该key的任意任务终端重启，或重新上线该操作会失效
     *
     * @param key
     */
    public void removeWorker(String key){
        simpleRedisConfig.getRedisInterface().delete(ScheduledOri + key);
        simpleRedisConfig.getRedisInterface().delete(ScheduledTig + key);

    }



//----------------------------------------------------------------------------------------------


    private Long getSchDur(JSONObject jsonObject){

        //优先 rate
        Long rate=jsonObject.getLong("rate",0L);

        if(rate>0L) {
            return rate;
        }

        return getSchDurWithOutRate(jsonObject);
    };


    private Long getSchDurWithOutRate(JSONObject jsonObject){

        //优先 delay
        Long fixedDelay=jsonObject.getLong("delay",0L);

        if(fixedDelay>0L) {
            return fixedDelay;
        }

        //其次cron 算出下一个时间
        String corn = jsonObject.getStr("corn");
        if(corn==null){
            throw  new RuntimeException("解析时间错误"+jsonObject);
        }

        TimeZone timeZone;

        String zone = jsonObject.getStr("zone");

        if (zone!=null) {
            timeZone = parseTimeZoneString(zone);
        }
        else {
            timeZone = TimeZone.getDefault();
        }


        return getNextTime(corn, timeZone).getTime()-System.currentTimeMillis();

    };



    private Date getNextTime(String cron, TimeZone timeZone){
        String key=cron+timeZone.toString();

        CronSequenceGenerator cronSequenceGenerator = cronSequenceGeneratorHashMap.get(key);

        if(cronSequenceGenerator==null){
            cronSequenceGenerator=new CronSequenceGenerator(cron,timeZone);
            cronSequenceGeneratorHashMap.put(key,cronSequenceGenerator);
        }
        return cronSequenceGenerator.next(new Date());

    };

    public static TimeZone parseTimeZoneString(String timeZoneString) {
        TimeZone timeZone = TimeZone.getTimeZone(timeZoneString);
        if ("GMT".equals(timeZone.getID()) && !timeZoneString.startsWith("GMT")) {
            throw new IllegalArgumentException("Invalid time zone specification '" + timeZoneString + "'");
        }
        return timeZone;
    }

}
